package com.teemor;

import java.util.concurrent.Executor;

/**
 * @author zhoulike
 * @date 2024/7/9 14:29
 */

public class Node {

    /**
     * 1 alive
     * 0 dead
     * 2 pending
     */
    private int state;

    private int id;

    private int term = 0;

    /**
     * 1 leader
     * 2 follower
     * 3 candidate
     */
    private int role;

    /**
     * 收到的投票
     */
    private int vote = 0;


    private int nodeCount;


    private long timeout = 1000 + (long) (Math.random() * 1000);

    private Channel channel;
    private long lastHeartbeatTime = System.currentTimeMillis();

    private Executor executor;

    public int getId() {
        return id;
    }

    private Logger logger;

    public Node(Channel channel, int nodeCount, Executor executor) {
        this.channel = channel;
        this.nodeCount = nodeCount;
        this.id = (int) (Math.random() * 1000);
        this.role = 2;
        this.executor = executor;

        this.logger = new Logger(id);

    }

    public void connect() {
        channel.connect(this);

    }


    public void restart() {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        channel.connect(this);
        // 重启后先是pending状态,需要等到同步master消息心跳后,才变为启动状态
        this.state = 2;
        this.term = 0;
        this.role = 2;
        this.lastHeartbeatTime = System.currentTimeMillis();
        logger.log("重启节点: id=%s", id);
    }


    public void onMessage(Message message) {
        switch (message.getType()) {
            case HEARTBEAT:
                // 收到心跳信息, 说明master节点选出来了
                this.lastHeartbeatTime = message.getTimestamp();
                this.role = 2;
                if (this.term > (int) message.getData()) {
                    logger.log("收到心跳异常信息: sendId=%s, masterId=%s, cTerm=%s, term=%s",
                            message.getSendId(), message.getSendId(), term, message.getData());
                }


                this.term = (int) message.getData();

                if (this.state == 2) {
                    this.state = 1;
                    logger.log("重启后同步数据: masterId=%s, term=%s", message.getSendId(), term);
                }
//                Logger.log("收到心跳信息: sendId=%s, term=%s", message.getSendId(), term);
                break;
            case ELECTION:
                doElection(message);
                break;
            case ELECTION_RESULT:
                doElectionResult(message);
                break;
            default:
        }

    }


    /**
     * 投票状态: 1 未投票 0 已投票
     */
    private int electionState;

    public void doElection(Message message) {
        this.role = 3;

        Message.ElectionMessage electionMessage = (Message.ElectionMessage) message.getData();
        logger.log("收到选举信息: sendId=%s, cTerm=%s, term=%s", message.getSendId(), term, electionMessage.getTerm());

        if (this.term == electionMessage.getTerm()) {
            // 如果是同一轮选举，则发送投票结果
            channel.sendMessage(new Message(Message.Type.ELECTION_RESULT, this.id, message.getSendId(),
                    new Message.ElectionMessage(this.term, electionState)));
        } else if (this.term < electionMessage.getTerm()) {
            // 如果是新一轮选举，则更新term，并发送投票结果
            this.term = electionMessage.getTerm();
            channel.sendMessage(new Message(Message.Type.ELECTION_RESULT, this.id, message.getSendId(),
                    new Message.ElectionMessage(this.term, 1)));
        }
    }

    public void doElectionResult(Message message) {
        Message.ElectionMessage electionMessage = (Message.ElectionMessage) message.getData();

        // 如果是同一轮选举，则更新投票状态
        if (electionMessage.getTerm() == this.term) {
            if (electionMessage.getResult() == 1) {
                this.vote++;

                // 如果投票数超过半数，则成为leader
                if (this.vote > this.nodeCount / 2) {
                    this.role = 1;
                    logger.log("成为leader, id=%s, term=%s, vote=%s", this.id, this.term, this.vote);
                    // 发送心跳消息
                    heartbeat();
                }
            }
        }


    }

    public void heartbeat() {
        channel.getNodes().stream()
                .filter(nodeId -> nodeId != this.id)
                .forEach(nodeId -> channel.sendMessage(new Message(Message.Type.HEARTBEAT, this.id, nodeId, this.term
                )));
    }

    public void election() {
        logger.log("发起选举, id=%s, term=%s", this.id, this.term);
        this.term++;
        this.role = 3;

        // 自己优先投票给自己
        this.electionState = 0;
        this.vote = 1;

        channel.getNodes().stream()
                .filter(nodeId -> nodeId != this.id)
                .forEach(nodeId -> channel.sendMessage(new Message(Message.Type.ELECTION, this.id, nodeId,
                        new Message.ElectionMessage(
                                this.term, 0)
                )));

    }

    private long electionTimeout = 1000 + (long) (Math.random() * 1000);

    private long lastElectionTime;

    public void start() {
        executor.execute(() -> {
            this.state = 1;
            connect();
            int loop = 200;
            while (true) {
                try {
                    Thread.sleep(100);
                    if (this.state == 1) {
                        // 最近一次心跳超时
                        if (System.currentTimeMillis() - this.lastHeartbeatTime > this.timeout && this.role == 2) {
                            // 状态改为候选者
                            this.role = 3;
                            // 发起选举
                            election();
                        }

                        // 选举超时，开启下一轮选举
                        if (System.currentTimeMillis() - this.lastElectionTime > this.electionTimeout && this.role == 3) {
                            // 发起选举
                            election();
                        }

                        if (this.role == 1) {
                            // 当心跳超过一定次数后, master节点重启, 以模拟故障
                            if (loop-- <= 0) {
                                this.restart();
                                continue;
                            }
                            heartbeat();
                        }
                    }

                    if (state == 2) {
                        // 最近一次心跳超时
                        if (System.currentTimeMillis() - this.lastHeartbeatTime > this.timeout && this.role == 2) {
                            // 状态改为候选者
                            this.role = 3;
                            // 发起选举
                            election();
                        }
                    }

                } catch (InterruptedException ig) {

                }
            }
        });

    }
}
